package com.nokia.ion.sdu.cdr_decoder;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class AnalysisWriter extends CdrWriter {
    private HashMap<String, Worker> workerMap = new HashMap<>();
    private LinkedList<TreeSet<Cdr>> results = new LinkedList<>();
    private ExecutorService es = Executors.newCachedThreadPool();

    private CsvExporter csvExporter;

    private Configuration.AnalysisConfigure analysisConfigure;

    public AnalysisWriter(LinkedBlockingQueue<Cdr> queue, Configuration configuration) throws IOException {
        super(queue, configuration);

        this.analysisConfigure = configuration.analysis;

        csvExporter = new CsvExporter(new File(analysisConfigure.outputPath, analysisConfigure.csvFilename));
    }

    @Override
    public void write(Cdr cdr) throws InterruptedException {
        String nodeId = cdr.nodeID;
        Worker worker = workerMap.get(nodeId);
        if (worker == null) {
            worker = new Worker(nodeId);
            workerMap.put(nodeId, worker);

            es.execute(worker);
        }
        worker.put(cdr);
    }

    @Override
    void close() {
        Cdr cdr = new Cdr();
        cdr.last = true;

        System.out.println("Number of worker " + workerMap.size());
        for (Worker t : workerMap.values()) {
            try {
                t.put(cdr);
            } catch (InterruptedException e) {
                System.err.println(e);
            }
        }

        try {
            es.shutdown();
            es.awaitTermination(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            es.shutdownNow();
        }

        try {
            csvExporter.newHeader();
            for (TreeSet<Cdr> ts : results) {
                csvExporter.newLine();
                for (Cdr c : ts) {
                    System.out.println(c.toString());
                    System.out.println();
                    csvExporter.write(c);
                }
            }
        } catch (IOException e) {
            System.err.println(e);
        } finally {
            csvExporter.close();
        }
    }

    class Worker extends Thread {
        private final String[] Exclude_A = {
                "220.206.133",
                "220.206.135",
                "220.206.137",
                "220.206.167",
                "220.206.169",
                "220.206.171",
        };

        private final String[] Exclude_B = {
                "116.79.201",
                "116.79.202",
                "116.79.217",
                "116.79.239",
        };

        private final String[] Liaoning_Addr = {
                "220.206.136",
                "116.79.220",
                "220.206.170",
                "116.79.240"
        };

        private static final int SGSN_LIAONING = 1;
        private static final int SGSN_NEIGHBOUR = 2;
        private static final int SGSN_OTHER = 3;

        private String name;
        private LinkedBlockingQueue<Cdr> queue = new LinkedBlockingQueue<>(1000);
        private HashMap<String, TreeSet<Cdr>> data = new HashMap<>();

        Worker(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            Cdr cdr;
            try {
                while (true) {
                    cdr = queue.take();
                    if (cdr.last) {
                        break;
                    }

                    cdr.sgsnLoc = getSgsnLoc(cdr.servingNodeAddress);
                    String key = String.format("%s%d%d", cdr.nodeID ,cdr.servedIMSI ,cdr.chargingId);
                    TreeSet<Cdr> cdrTree = data.get(key);
                    if (cdrTree == null) {
                        cdrTree = new TreeSet<>((o1, o2) -> o1.recordOpeningTime.compareTo(o2.recordOpeningTime));
                        data.put(key, cdrTree);
                    }
                    cdrTree.add(cdr);
                }

                for (Map.Entry<String, TreeSet<Cdr>> entry : data.entrySet()) {
                    TreeSet<Cdr> ts = entry.getValue();

                    Cdr first = ts.first();
                    Cdr second = ts.higher(first);
                    while (second != null) {
                        if (first.sgsnLoc == SGSN_LIAONING && second.sgsnLoc == SGSN_OTHER) {
                            results.add(ts);
                            break;
                        }
                        first = second;
                        second = ts.higher(first);
                    }
                }
            } catch (InterruptedException e) {
                System.err.println(e);
                System.exit(-1);
            }
        }

        public void put(Cdr cdr) throws InterruptedException {
            queue.put(cdr);
        }

        private boolean isLiaoning(String ip) {
            for (String addr : Liaoning_Addr) {
                if (ip.startsWith(addr)) {
                    return true;
                }
            }

            return false;
        }

        private boolean isNeighbour(String ip) {
            if (ip.startsWith("220.206")) {
                for (String addr : Exclude_A) {
                    if (ip.startsWith(addr)) {
                        return true;
                    }
                }
            } else if (ip.startsWith("116.79")) {
                for (String addr : Exclude_B) {
                    if (ip.startsWith(addr)) {
                        return true;
                    }
                }
            }

            return false;
        }

        private byte getSgsnLoc(String ip) {
            if (isLiaoning(ip)) {
                return SGSN_LIAONING;
            }

            if (isNeighbour(ip)) {
                return SGSN_NEIGHBOUR;
            }

            return SGSN_OTHER;
        }
    }
}


